package com.qyb.mq.consumer;

import com.qyb.util.JsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * *************************************************************************************************
 * <p/>
 * 实现功能：
 * <p>
 * ------------------------------------------------------------------------------------------------
 * 版本          变更时间             变更人                     变更原因
 * ------------------------------------------------------------------------------------------------
 * 1.0.00      2017/7/17 13:44      秦有宝                     新建
 * <p/>
 * *************************************************************************************************
 */
public class MqConsumer<M> implements Runnable{
    /**
     * 消息的类型(生产和消费保持一致)
     */
    private String topic;
    //kafka参数配置
    private static final String MQ_CONFIG_KEY="MQ_CONSUMER_CONFIG";
    private static Properties kafkaProperties;
    private Class<M> messageClass;

    //缓冲区大小(默认=1，无缓冲)
    private int bufferSize;

    /**
     * 获取缓冲区大小
     * @return 缓冲区大小
     */
    public int getBufferSize() {
        return bufferSize;
    }

    /**
     * 设置缓冲区大小
     * @param bufferSize 缓冲区大小
     */
    public void setBufferSize(int bufferSize) {
        this.bufferSize = bufferSize;
    }

    /**
     * 构造函数
     * @param topic 消息的类型
     * @param groupId 消费者分组
     * @param messageClass 业务实体类型
     */
    public MqConsumer(String topic, String groupId, Class<M> messageClass){
        this.topic = topic;
//        kafkaProperties = GlobalParam.getProperties(MQ_CONFIG_KEY);
//        if (kafkaProperties==null){
//            throw new ExceptionInInitializerError("系统没有配置kafka的消费者全局参数："+MQ_CONFIG_KEY);
//        }
        kafkaProperties.setProperty("bootstrap.servers","10.4.68.45:9092,10.4.68.46:9092");
        kafkaProperties.setProperty("enable.auto.commit","false");
        kafkaProperties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProperties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        kafkaProperties.setProperty("group.id",groupId);
        this.messageClass = messageClass;
        this.bufferSize = 1;
    }

    /**
     * 创建一个消息消费者
     * @return 消息消费者
     */
    private KafkaConsumer<String,String> createConsumer(){
        return new KafkaConsumer<>(kafkaProperties);
    }

    /**
     * 处理消息的业务逻辑
     * @param key 消息键值
     * @param message 消息
     */
    public void processMessage(String key,M message){
        //业务逻辑处理
    }

    /**
     * 将JSON字符串反序列化为业务实体
     * @param json json字符串
     * @return 业务实体
     */
    private M toEntity(String json){
        try {
            return JsonUtils.fromJson(json,messageClass);
        } catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }
    /**
     * 处理队列消息记录
     * @param records 消息记录
     */
    private void processRecords(List<ConsumerRecord<String, String>> records){
        records.forEach((r)->processMessage(r.key(),toEntity(r.value())));
    }
    /**
     * 开始轮询处理消息队列
     */
    public final void startPoll(){
        KafkaConsumer<String,String> consumer = createConsumer();
        consumer.subscribe(Collections.singletonList(topic));
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        //轮询
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(100);
            if (records.count()>0){
                records.forEach(buffer::add);
            }
            if (buffer.size()>=bufferSize){
                //处理消息
                processRecords(buffer);
                //提交偏移量,清除缓存
                consumer.commitSync();
                buffer.clear();
            }
        }
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        //开始轮询处理消息队列
        startPoll();
    }
}
